package com.imooc.rabbit.producer.broker;

import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import com.imooc.rabbit.api.Message;
import com.imooc.rabbit.api.MessageType;
import com.imooc.rabbit.api.exception.MessageRunTimeException;
import com.imooc.rabbit.common.convert.GenericMessageConverter;
import com.imooc.rabbit.common.convert.RabbitMessageConverter;
import com.imooc.rabbit.common.serializer.Serializer;
import com.imooc.rabbit.common.serializer.SerializerFactory;
import com.imooc.rabbit.common.serializer.impl.JacksonSerializerFactory;
import com.imooc.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * @description: rabbitTemplate池化封装
 *  每一个topic，对应一个rabbitTemplate
 *      1.提高发送效率
 *      2.可以根据不同的需求制定化不同的rabbitTemplate，比如每一个topic 都有自己的routingKey规则
 * @author: Yz
 * @create: 2020-10-10 13:48
 **/
@Component
@Slf4j
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private ConnectionFactory connectionFactory;

    @Autowired
    private MessageStoreService messageStoreService;

    private SerializerFactory serializerFactory = JacksonSerializerFactory.INSTANCE;

    /**
     * 池化封装的Map
     */
    private Map<String, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    private Splitter splitter = Splitter.on("#");

    public RabbitTemplate getTemplate(Message message) throws MessageRunTimeException {
        //判空
        Preconditions.checkNotNull(message);
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);
        if (rabbitTemplate != null) {
            return rabbitTemplate;
        }
        log.info("#RabbitTemplateContainer.getTemplate# topic:{} is not exists,create one", topic);

        RabbitTemplate newTemplate = new RabbitTemplate(connectionFactory);
        newTemplate.setExchange(topic);
        newTemplate.setRoutingKey(message.getRoutingKey());
        //重试次数 默认
        newTemplate.setRetryTemplate(new RetryTemplate());
        //添加序列化反序列化converter对象
        Serializer serializer = serializerFactory.create();
        GenericMessageConverter gmc = new GenericMessageConverter(serializer);
        RabbitMessageConverter rmc = new RabbitMessageConverter(gmc);
        //对于message的序列化方式
        newTemplate.setMessageConverter(rmc);
        String messageType = message.getMessageType();
        //除了迅速消息，都要确认comfirmCallBack
        if (!MessageType.RAPID.equals(messageType)) {
            newTemplate.setConfirmCallback(this);
        }
        rabbitMap.putIfAbsent(topic, newTemplate);
        return rabbitMap.get(topic);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        //具体的消息应答
        List<String> strings = splitter.splitToList(correlationData.getId());
        String messageId = strings.get(0);
        long sendTimes = Long.parseLong(strings.get(1));
        //true成功返回结果
        if (ack) {
            //当broker 返回ack成功时，就是更新一下日志表里对应的状态为send_ok
            messageStoreService.success(messageId);
            log.info("#RabbitTemplateContainer.confirm# send message is OK,confirm messageId: {},sendTime: {}", messageId, sendTimes);
        } else {
            log.error("#RabbitTemplateContainer.confirm# send message is Fail,confirm messageId: {},sendTime: {}", messageId, sendTimes);
        }
    }
}
